package com.sql.blood.spark

import com.sql.blood.common.util.GraphDatabaseConnection
import org.apache.spark.sql.execution.QueryExecution
import org.apache.spark.sql.util.QueryExecutionListener

class SparkSqlListener extends QueryExecutionListener with Logging {

  override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit = {
    val sparkSqlParser = new SparkSqlParser()
    val output = sparkSqlParser.resolveLogicPlan(qe.sparkSession, qe.logical)
    val inputTables = output._1
    val outputTables = output._2
    val deleteTables = output._3

    println("---inputTables---")
    inputTables.foreach(x=>println(x.toString))
    println("---outputTables---")
    outputTables.foreach(x=>println(x.toString))
    println("---deleteTables---")
    deleteTables.foreach(x=>println(x.toString))

    val conn = new GraphDatabaseConnection()
    inputTables.foreach{input =>
      conn.insertVertex(input)
    }
    outputTables.foreach{output =>
      conn.insertVertex(output)
    }
    inputTables.foreach{input =>
      outputTables.foreach{output =>
        conn.insertEdge(input, output)
      }
    }
    conn.close();
  }

  override def onFailure(funcName: String, qe: QueryExecution, exception: Exception): Unit = {}
}
